package com.wuwangfu.func;

import com.wuwangfu.entity.ActivityBean;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava30.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava30.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Description：
 * @Author：jcshen
 * @Date：2023-07-04
 */
public class ActivityCountFunc extends KeyedProcessFunction<Tuple2<String,String>, ActivityBean,ActivityBean> {

    //按小时未去重的次数
    private transient MapState<String, Long> hourUserState;
    //按小时去重后的次数
    private transient MapState<String, Long> hourDisUserState;
    //判断按小时去重的布隆过滤器
    private transient MapState<String, BloomFilter<String>>  hourBloomFilterState;

    @Override
    public void open(Configuration parameters) throws Exception {
        /**
         * (2023-01-24 16->2000)
         * (2023-01-24 17->2000)
         */
        MapStateDescriptor<String, Long> hourStateDesc = new MapStateDescriptor<>("hour-user-count-state", String.class, Long.class);
        hourUserState = getRuntimeContext().getMapState(hourStateDesc);
        /**
         * (2023-01-24 16->1000)
         * (2023-01-24 17->1000)
         */
        MapStateDescriptor<String, Long> hourDistUserStateDesc = new MapStateDescriptor<>("hour-dist-user-count-state", String.class, Long.class);
        hourDisUserState = getRuntimeContext().getMapState(hourDistUserStateDesc);
        //状态TTL，设置存活时间
        StateTtlConfig hourBloomTtlConfig = StateTtlConfig.newBuilder(Time.minutes(90))//存活90分钟
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();

        //布隆过滤器
        MapStateDescriptor<String, BloomFilter<String>> hourBloomDesc = new MapStateDescriptor<String,BloomFilter<String>>(
                "hour-bloom-state",
                TypeInformation.of(String.class),
                TypeInformation.of(new TypeHint<BloomFilter<String>>() {}));

        hourBloomDesc.enableTimeToLive(hourBloomTtlConfig);
        hourBloomFilterState = getRuntimeContext().getMapState(hourBloomDesc);
    }

    @Override
    public void processElement(ActivityBean value, KeyedProcessFunction<Tuple2<String, String>, ActivityBean, ActivityBean>.Context ctx, Collector<ActivityBean> out) throws Exception {

        String uid = value.getUid();

        //为去重的次数
        String dayAndHour = value.getTime().substring(0, 13);
        Long hourCount = hourUserState.get(dayAndHour);
        if (hourCount == null){
            hourCount = 0L;
        }
        //累加
        hourCount += 1;
        //更新状态
        hourUserState.put(dayAndHour,hourCount);

        //去重后的次数
        BloomFilter<String> hourBloomFilter = hourBloomFilterState.get(dayAndHour);
        Long hourDisCount = hourDisUserState.get(dayAndHour);
        if (hourBloomFilter == null){
            hourBloomFilter = BloomFilter.create(Funnels.unencodedCharsFunnel(),100000);
            hourDisCount = 0L;
        }
        //判断用户在这个小时内是否存在
        if(!hourBloomFilter.mightContain(uid)){//不存在
            hourDisCount += 1L;
            hourBloomFilter.put(uid);
        }
        //更新去重后的状态
        hourDisUserState.put(dayAndHour,hourDisCount);
        //更新布隆过滤器
        hourBloomFilterState.put(dayAndHour,hourBloomFilter);
        //返回值
        value.setTime(dayAndHour);
        value.setPv(hourCount);
        value.setUv(hourDisCount);

        out.collect(value);
    }
}
